package com.wingtech.mqtt.client;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author haungfeilong@wingtech.com
 * @version 1.0
 * @ClassName MqttPushClient
 * @description mqtt客户端
 * @date 2020/11/26 17:38
 * @since JDK 1.8
 */
@Component
@Slf4j
public class MqttPushClient {

    @Autowired
    private PushCallback pushCallback;

    private MqttClient client;

    public MqttClient getClient() {
        return client;
    }

    public void setClient(MqttClient client) {
        this.client = client;
    }

    /**
     * connect : 客户端连接
     *
     * @author huangfeilong@wingtech.com
     * @version 1.0
     * @date 2020/11/26 17:59
     * @param host	ip端口
     * @param clientID	客户端Id
     * @param username	用户名
     * @param password	密码
     * @param timeout	超时时间
     * @param keepalive	保留位数
     * @return void
     * @since JDK 1.8
     */
    public void connect(String host, String clientID, String username, String password, int timeout, int keepalive){
        MqttClient client;
        try {
            client = new MqttClient(host, clientID, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(timeout);
            options.setKeepAliveInterval(keepalive);
            this.setClient(client);
            client.setCallback(pushCallback);
            client.connect(options);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * publish : 发布
     *
     * @author huangfeilong@wingtech.com
     * @version 1.0
     * @date 2020/11/26 18:33
     * @param qos	连接方式
     * @param retained	是否保留
     * @param topic	主题
     * @param pushMessage	消息体
     * @return void
     * @since JDK 1.8
     */
    public void publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        MqttTopic mTopic = this.getClient().getTopic(topic);
        if (null == mTopic) {
            log.error("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * subscribe : 订阅某个主题
     *
     * @author huangfeilong@wingtech.com
     * @version 1.0
     * @date 2020/11/26 18:34
     * @param topic
     * @param qos
     * @return void
     * @since JDK 1.8
     */
    public void subscribe(String topic, int qos) {
        log.info("开始订阅主题" + topic);
        try {
            this.getClient().subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}
